package com.atguigu.flink.chapter5.source;


import com.atguigu.flink.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.Socket;
import java.nio.charset.StandardCharsets;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2020/12/19 11:26
 */
public class Flink04_Source_Custom {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 自定义一个可以从socket读数据
        DataStreamSource<WaterSensor> ds = env.addSource(new MySocketSource("hadoop162", 9999));
        ds.print();

        env.execute();
    }


    public static class MySocketSource implements SourceFunction<WaterSensor> {
        private String host;
        private int port;
        private boolean cancel = false;
        private Socket socket;
        private BufferedReader reader;

        public MySocketSource(String host, int port) {
            this.host = host;
            this.port = port;
        }

        @Override
        public void run(SourceContext<WaterSensor> ctx) throws Exception {
            // 从socket读取  hadoop162 9999
            socket = new Socket(host, port);
            reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
            String line = reader.readLine();
            while (!cancel && line != null) {
                String[] split = line.split(",");
                ctx.collect(new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2])));
                line = reader.readLine();
            }
        }

        @Override
        public void cancel() {
            cancel = true;
            if (reader != null) {
                try {
                    reader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (socket != null) {
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

}
/*
sensor_1,1607527992000,20
 */